package com.zy.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.io.IOException;

public class KafkaUtil {
    public static final String bootStrap = "hadoop102:9092,hadoop103:9092,hadoop104:9092";


    public static String getUpsertKafkaSinkConnector(String topic){
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '"+ topic +"',\n" +
                "  'properties.bootstrap.servers' = 'hadoop102:9092,hadoop103:9092,hadoop104:9092',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }

    public static String getKafkaSourceTable(String groupId){
        return "CREATE TABLE topic_db (\n" +
                "  `database` STRING,\n" +
                "  `table` STRING,\n" +
                "  `type` STRING,\n" +
                "  `ts` STRING,\n" +
                "  `data` MAP<STRING,STRING>,\n" +
                "  `old` MAP<STRING,STRING>,\n" +
                "  proc_time AS PROCTIME()\n" +
                ")" + getKafkaSinkConnector("topic_db", groupId) ;
    }

    public static String getKafkaSinkConnector(String topic ,String groupId){
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '"+ topic +"',\n" +
                "  'properties.bootstrap.servers' = '"+ bootStrap +"',\n" +
                "  'properties.group.id' = '"+ groupId +" ',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    public static KafkaSink<String> getKafkaSink(String topic){

        //获取到输出到kafka主题的来连接器
        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers(bootStrap)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                //.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)//至少一次
                .build();
        //在生产环境中，如果要想保证写入的一致性，需要进行如下的日志
        //1.指定DeliveryGuarantee.EXACTLY_ONCE
        //2.setTransactionalIdPrefix("")
        //3.检查点开启
        //4.事务超时时间大于检查点超时时间  但是要小于事务最大超时时间(默认15min)
        //5.在消费端，对于没有提交的消息，不应该被消费
        // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
        // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
        return sink;
    }

    public static KafkaSource<String> getKafkaSource(String topic,String groupId){
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(bootStrap)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")//二次提交
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] bytes) throws IOException {
                        //防止null
                        if (bytes != null){
                            return new String(bytes);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String s) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                })
                .build();

        return kafkaSource;
    }

    public static <T>KafkaSink<T> getKafkaSinkBySchema(KafkaRecordSerializationSchema<T> kafkaRecordSerializationSchema){

        //获取到输出到kafka主题的来连接器
        KafkaSink<T> sink = KafkaSink.<T>builder()
                .setBootstrapServers(bootStrap)
                .setRecordSerializer(kafkaRecordSerializationSchema
                )
                //.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)//至少一次
                .build();
        //在生产环境中，如果要想保证写入的一致性，需要进行如下的日志
        //1.指定DeliveryGuarantee.EXACTLY_ONCE
        //2.setTransactionalIdPrefix("")
        //3.检查点开启
        //4.事务超时时间大于检查点超时时间  但是要小于事务最大超时时间(默认15min)
        //5.在消费端，对于没有提交的消息，不应该被消费
        // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
        // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
        return sink;
    }

}
